package com.study.zkyxgs1.utils;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.study.zkyxgs1.exception.ZkyxgsException;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.NestedQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.*;

/**
 * Es操作工具类
 * @author weixiaochun
 * @date 2022/6/16 7:07
 */
@Component("esDB")
@Slf4j
public class ESUtils<T> implements DBUtils<T> {

    @Resource
    private RestHighLevelClient restHighLevelClient;

    /**
     * 索引
     */
    @Value("${elasticsearch.index}")
    private String INDEX_NAME;

    /**
     * 滚动超时
     */
    private static final long SCROLL_TIMEOUT = 180000;

    /**
     * 滚动大小
     */
    private static final int SIZE = 1000;

    /**
     * 获取实体的id值
     * @param entity 实体
     * @return
     */
    private Object getIdValue(T entity) {
        try {
            Field field = entity.getClass().getDeclaredField("id");
            field.setAccessible(true);
            return field.get(entity) != null
                    ? field.get(entity)
                    : UUID.randomUUID().toString().replace("-", "");
        } catch (NoSuchFieldException | IllegalAccessException e) {
            e.printStackTrace();
            log.error("{} --- error get id", entity);
            throw new ZkyxgsException(ExceptionConstant.CLASS_EXCEPTION.getCode(), ExceptionConstant.CLASS_EXCEPTION.getMessage());
        }
    }

    private String toJsonString(T entity){
        try{
            ObjectMapper mapper = new ObjectMapper();
            return mapper.writeValueAsString(entity);
        }catch (IOException e){
            e.printStackTrace();
            log.error("{} --- error converting json", entity);
            throw new ZkyxgsException(ExceptionConstant.DATA_CONVERSION_EXCEPTION.getCode(), ExceptionConstant.DATA_CONVERSION_EXCEPTION.getMessage());
        }
    }

    private List<T> scrollQuery(SearchSourceBuilder builder, Class<T> clazz){
        List<T> list = new ArrayList<>();

        //滚动查询的Scroll
        Scroll scroll = new Scroll(TimeValue.timeValueMillis(SCROLL_TIMEOUT));

        SearchRequest request = new SearchRequest(INDEX_NAME);

        // 构建大小
        builder.size(SIZE);

        request.scroll(scroll);
        request.source(builder);

        //存储scroll的list
        List<String> scrollIdList = new ArrayList<>();

        ObjectMapper mapper = new ObjectMapper();
        try{
            //返回结果
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            //拿到第一个ScrollId（游标）
            String scrollId = searchResponse.getScrollId();
            //拿到hits结果
            SearchHits hits = searchResponse.getHits();
            scrollIdList.add(scrollId);
            //滚动查询将SearchHit封装到result中
            while (hits != null && hits.getHits().length > 0) {
                for (SearchHit hit : hits) {
                    //Function<SearchHit, T>, 输入SearchHit，经过操作后，返回T结果
                    list.add(mapper.readValue(hit.getSourceAsString(), clazz));
                }
                //说明滚动完了，返回结果即可
                if (hits.getHits().length < 1000) {
                    break;
                }
                //继续滚动，根据上一个游标，得到这次开始查询位置
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                //得到结果
                SearchResponse searchScrollResponse = restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
                //定位游标
                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits();
                scrollIdList.add(scrollId);
            }


        }catch (IOException e){
            e.printStackTrace();
            log.error("query error\nDSL: {}", request.source().toString());
            throw new ZkyxgsException(ExceptionConstant.DATA_QUERY_EXCEPTION.getCode(), ExceptionConstant.DATA_QUERY_EXCEPTION.getMessage());
        }finally {
            //清理scroll,释放资源
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.setScrollIds(scrollIdList);
            try {
                restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
                log.error("error releasing resource");
            }
        }

        return list;
    }


    /**
     * 保存单条数据
     *
     * @param data 保存的数据
     */
    @Override
    public void save(T data) {
        if(data == null){
            log.error("data is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        IndexRequest request = new IndexRequest(INDEX_NAME);
        String jsonData = null;
        jsonData = toJsonString(data);
        request.id(getIdValue(data).toString()).source(jsonData, XContentType.JSON);
        try {
            restHighLevelClient.index(request, RequestOptions.DEFAULT);
        }catch (IOException e){
            e.printStackTrace();
            log.error("{} --- save the error", data);
            throw new ZkyxgsException(ExceptionConstant.DATA_SAVE_EXCEPTION.getCode(), ExceptionConstant.DATA_SAVE_EXCEPTION.getMessage());
        }

    }

    /**
     * 批量保存
     *
     * @param entityList 保存的数据集合
     */
    @Override
    public void saveBatch(Collection<T> entityList) {
        if(entityList == null || entityList.size() == 0){
            log.error("entityList is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        BulkRequest request = new BulkRequest(INDEX_NAME);
        for (T data : entityList) {
            request.add(new IndexRequest().id(getIdValue(data).toString())
            .source(toJsonString(data), XContentType.JSON));
        }
        try{
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        }catch (IOException e){
            e.printStackTrace();
            log.error("{} --- batch save the error", entityList);
            throw new ZkyxgsException(ExceptionConstant.DATA_SAVE_EXCEPTION.getCode(), ExceptionConstant.DATA_SAVE_EXCEPTION.getMessage());
        }
    }

    /**
     * 根据id删除数据
     *
     * @param id    id
     * @param clazz 指定的class
     */
    @Override
    public boolean removeById(Object id, Class<T> clazz) {
        if(ObjectUtils.isEmpty(id)){
            log.error("id is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        DeleteByQueryRequest request = new DeleteByQueryRequest(INDEX_NAME);
        request.setQuery(new TermQueryBuilder("id", id));
        try{
            BulkByScrollResponse response = restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
            return response.getDeleted() > 0;
        }catch (IOException e){
            e.printStackTrace();
            log.error("id:{} --- delete the error", id);
            throw new ZkyxgsException(ExceptionConstant.DATA_DELETE_EXCEPTION.getCode(), ExceptionConstant.DATA_DELETE_EXCEPTION.getMessage());
        }
    }

    /**
     * 根据map条件删除数据
     *
     * @param map   map条件
     * @param clazz 指定class
     */
    @Override
    public void removeByMap(Map<String, Object> map, Class<T> clazz) {
        if(map == null || map.size() == 0){
            log.error("map is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        DeleteByQueryRequest request = new DeleteByQueryRequest(INDEX_NAME);

        for (String key : map.keySet()) {
            request.setQuery(new TermQueryBuilder(key, map.get(key)));
        }

        try{
            restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
        }catch (IOException e){
            e.printStackTrace();
            log.error("query data:{} --- batch delete the error", map);
            throw new ZkyxgsException(ExceptionConstant.DATA_DELETE_EXCEPTION.getCode(), ExceptionConstant.DATA_DELETE_EXCEPTION.getMessage());
        }

    }

    /**
     * 根据id更新数据
     *
     * @param data 更新数据
     * @return
     */
    @Override
    public boolean updateById(T data) {
        if(data == null){
            log.error("data is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        UpdateRequest request = new UpdateRequest();
        request.index(INDEX_NAME)
                .id(getIdValue(data).toString())
                .doc(toJsonString(data), XContentType.JSON);

        try{
            UpdateResponse response = restHighLevelClient.update(request, RequestOptions.DEFAULT);
            return true;
        }catch (IOException e){
            e.printStackTrace();
            log.error("data:{} --- update the error", data);
            throw new ZkyxgsException(ExceptionConstant.DATA_UPDATE_EXCEPTION.getCode(), ExceptionConstant.DATA_UPDATE_EXCEPTION.getMessage());
        }
    }

    /**
     * 批量根据id更新数据
     *
     * @param list 更新数据集合
     */
    @Override
    public void updateBatchById(Collection<T> list) {
        list.forEach(this::updateById);
    }

    /**
     * 根据id查询数据
     *
     * @param id    id
     * @param clazz 指定的class
     * @return
     */
    @Override
    public T getById(Object id, Class<T> clazz) {
        if(ObjectUtils.isEmpty(id)){
            log.error("id is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchRequest request = new SearchRequest(INDEX_NAME);
        GetRequest getRequest = new GetRequest(INDEX_NAME, id.toString());
        request.source();
        try{
            GetResponse documentFields = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            ObjectMapper mapper = new ObjectMapper();
            return mapper.convertValue(documentFields.getSource(), clazz);
        }catch (IOException e){
            e.printStackTrace();
            log.error("ID:{} --- error obtaining data based on ID", id.toString());
            throw new ZkyxgsException(ExceptionConstant.DATA_QUERY_EXCEPTION.getCode(), ExceptionConstant.DATA_QUERY_EXCEPTION.getMessage());
        }
    }

    /**
     * 查询所有
     *
     * @param clazz 指定class
     * @return
     */
    @Override
    public List<T> getAll(Class<T> clazz) {
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.query(QueryBuilders.matchAllQuery());
        return scrollQuery(builder, clazz);
    }

    /**
     * 根据单个条件查询数据
     *
     * @param key   数据的key
     * @param val   数据的val
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByOneQuery(String key, Object val, Class<T> clazz) {
        if(StringUtils.isEmpty(key)){
            log.error("key is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.query(QueryBuilders.matchQuery(key, val));
        return scrollQuery(builder, clazz);
    }

    /**
     * 根据实体的数据进行查询
     *
     * @param data  实体类
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByEntity(T data, Class<T> clazz) {
        if(data == null){
            log.error("data is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchSourceBuilder builder = new SearchSourceBuilder();
        try{
            for (Field field : data.getClass().getDeclaredFields()) {
                field.setAccessible(true);
                if(field.get(data) != null){
                    builder.query(QueryBuilders.termQuery(field.getName(), field.get(data)));
                }
            }
            return scrollQuery(builder, clazz);
        }catch (IllegalAccessException e){
            e.printStackTrace();
            log.error("query data error\ndata: {}", data);
            throw new ZkyxgsException(ExceptionConstant.CLASS_EXCEPTION.getCode(), ExceptionConstant.CLASS_EXCEPTION.getMessage());
        }
    }

    /**
     * 根据范围查询数据
     *
     * @param key   条件的key
     * @param start 开始的值
     * @param end   结束的值
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByRange(String key, Object start, Object end, Class<T> clazz) {
        if(StringUtils.isEmpty(key)){
            log.error("key is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.postFilter(QueryBuilders.rangeQuery(key).gte(start).lte(end));
        return scrollQuery(builder, clazz);
    }

    /**
     * 根据集合查询数据
     *
     * @param key   条件的key
     * @param list  条件的值集合
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByList(String key, Collection list, Class<T> clazz) {
        if(StringUtils.isEmpty(key)){
            log.error("key is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchSourceBuilder builder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        for (Object o : list) {
            boolQueryBuilder.should(QueryBuilders.matchQuery(key, o));
        }
        builder.query(boolQueryBuilder);
        return scrollQuery(builder, clazz);
    }

    /**
     * 根据嵌套参数查询数据
     *
     * @param keys  嵌套条件的keys
     * @param val   条件的val
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByNest(List<String> keys, Object val, Class<T> clazz) {
        if(keys == null || keys.size() == 0){
            log.error("keys is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        SearchSourceBuilder builder = new SearchSourceBuilder();

        StringBuffer sb = new StringBuffer();
        for (String key : keys) {
            sb.append(".").append(key);
        }
        String key = sb.substring(1);
        NestedQueryBuilder nestedQueryBuilder = QueryBuilders
                .nestedQuery("description", QueryBuilders.matchQuery(key, val),
                        ScoreMode.None);
        builder.query(nestedQueryBuilder);
        return scrollQuery(builder, clazz);
    }

    /**
     * 根据嵌套参数查询数据
     * @param map 条件val
     * @param clazz 指定的class
     * @return
     */
    @Override
    public List<T> getByNest(Map<String, Object> map, Class<T> clazz) {
        if(map == null || map.size() == 0){
            log.error("map is null");
            throw new ZkyxgsException(ExceptionConstant.DATA_EXCEPTION.getCode(), ExceptionConstant.DATA_EXCEPTION.getMessage());
        }
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        SearchSourceBuilder builder = new SearchSourceBuilder();

        for (String key : map.keySet()) {
            StringBuffer sb = new StringBuffer();
            sb.append(".").append(key);
            joint(map, key, sb, boolQueryBuilder);
        }
        builder.query(boolQueryBuilder);
        return scrollQuery(builder, clazz);
    }

    /**
     *
     * @param map 当前map
     * @param key 当前map中的key
     * @param sb 嵌套key拼接字符串
     * @param boolQueryBuilder 多条件构造器
     */
    public void joint(Map map, String key, StringBuffer sb, BoolQueryBuilder boolQueryBuilder){
        if(map.get(key) instanceof Map){
            Map m = (Map) map.get(key);
            for (Object k : m.keySet()) {
                sb.append(".").append(k);
                joint(m, k.toString(), sb, boolQueryBuilder);
                sb.delete(sb.indexOf(k.toString()) - 1, sb.length());
            }
        }else{
            if(sb.lastIndexOf(".") == 0){
                boolQueryBuilder.should(QueryBuilders.termQuery(sb.substring(1), map.get(key)));
            }else{
                NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery(sb.toString().split("\\.")[1],
                        QueryBuilders.matchQuery(sb.substring(1), map.get(key)), ScoreMode.None);
                boolQueryBuilder.should(nestedQueryBuilder);
            }
        }
    }
}
